package com.such.kit.io.source;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import javax.sql.DataSource;
import org.apache.commons.collections4.multimap.ArrayListValuedHashMap;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import com.such.kit.Callable;
import com.such.kit.Logger;
import com.such.kit.bean.BeanCommons;
import com.such.kit.cache.CacheCommons;
import com.such.kit.database.MetaDataCommons;
import com.such.kit.database.dialect.DialectCommons;
import com.such.kit.exception.ExceptionCommons;
import com.such.kit.io.simple.support.IOSimple;
import com.such.kit.io.support.IOSource;
import com.such.kit.io.support.IOSqlProvider;
import com.such.kit.message.bean.MessageItem;
import com.such.kit.message.bean.Messager;
import com.such.kit.placeholder.PlaceholderCommons;
import com.such.kit.string.StringCommons;
import com.such.kit.validate.ValidateCommons;

/**
 * @author SUCH
 * <pre>
 * [继承扩展于 {@link IOSource}] 以数据库作为 IO 数据源的实现
 * 
 * {@link doInput} 方法 inputExpression 参数的推荐值为 insert 语句
 * [举例] insert into TABLE (id, name) values (:id, :name)
 * 
 * {@link doOutput} 方法 outputExpression 参数的推荐值为 select 语句
 * [举例] select id, name from TABLE where id = :id
 * 
 * {@link IOSqlProvider} 是动态 sql 语句供应器，用于处理动态、复杂的 sql 语句需求
 * 字符串格式为 #[SP]#[IOSqlProvider 实现类 className]#[sqlKey]
 * 使用时将字符串赋值于 inputExpression、outputExpression
 * 
 * [警告] 由于 DatabaseIOSource 的实现特性，只能使用字段名的方式进行值绑定引用数据、{@link IOSimple#args 自定义参数集}
 * 
 * {@link IOSimple#args 自定义参数集} 可以用作 sql 语句值绑定
 * </pre>
 */
public class DatabaseIOSource extends IOSource {

	/** EXPRESSION_SQL_PROVIDER：使用 {@link IOSqlProvider} 构建 sql 语句的表达式前缀 */
	public static final String EXPRESSION_SQL_PROVIDER = "SP";
	/** SQL_RESOURCE：[导入时] 的内置特殊表达式，表示为直接执行的 sql 语句 */
	public static final String SQL_RESOURCE = "sqlResource";

	private static String providerCahce = CacheCommons.buildCacheName(DatabaseIOSource.class);

	private ArrayListValuedHashMap<String, Object> inputResult = new ArrayListValuedHashMap<String, Object>();
	private boolean accuratePositioning = true;
	private JdbcTemplate jdbcTemplate;
	private NamedParameterJdbcTemplate namedParameterJdbcTemplate;
	private TransactionTemplate transactionTemplate;
	private DialectCommons dialectCommons;
	/** args：自定义参数集 */
	private Map<String, Object> args = new HashMap<String, Object>();

	/**
	 * @param dataSource 数据源
	 * @throws Exception
	 */
	public DatabaseIOSource(DataSource dataSource) throws Exception {
		this.jdbcTemplate = new JdbcTemplate(dataSource);
		this.namedParameterJdbcTemplate = new NamedParameterJdbcTemplate(this.jdbcTemplate);
		this.transactionTemplate = new TransactionTemplate(new DataSourceTransactionManager(dataSource));
		this.dialectCommons = DialectCommons.getInstance(MetaDataCommons.getDbType(dataSource));
	}

	/**
	 * <pre>
	 * 设置 {@link #args 自定义参数}
	 * </pre>
	 * @param key 参数键
	 * @param value 参数值
	 */
	public void putArg(String key, Object value) {
		this.args.put(key, value);
	}

	/**
	 * <pre>
	 * 设置自定义参数集
	 * </pre>
	 * @param args 参数集 [参考 {@link #args}]
	 */
	public void putArgs(Map<String, Object> args) {
		this.args.putAll(args);
	}

	/**
	 * <pre>
	 * 清空自定义参数集
	 * </pre>
	 */
	public void clearArgs() {
		this.args.clear();
	}

	/**
	 * <pre>
	 * [导入时] 是否精确定位模式
	 * 开启后，当批量执行异常则会进入逐条执行模式，此模式下可以定位到具体那条 sql 执行异常
	 * 
	 * [警告] 开启此模式会导致极端情况下效率低下
	 * </pre>
	 * @param accuratePositioning 是否精确定位模式 [默认值为 true] [true 为开启 | false 为关闭]
	 */
	public void openAccuratePositioning(boolean accuratePositioning) {
		this.accuratePositioning = accuratePositioning;
	}

	@Override
	public void doInput(String sql, Map<String, Object> params, boolean lastBatch, Map<String, Object> extraParams) {
		try {
			if (ValidateCommons.isNotEmpty(sql) && ValidateCommons.isNotEmpty(params)) {
				if (SQL_RESOURCE.equals(sql)) {
					this.inputResult.put(SQL_RESOURCE, params.get(SQL_RESOURCE));
				} else {
					params.putAll(this.args);
					autoReplace(params);
					sql = getSql(sql, params);
					this.inputResult.put(sql, new HashMap<String, Object>(params)); // put 新的对象，避免 put 引用导致的数据污染
				}
			}
			if (lastBatch) {
				Iterator<Entry<String, Collection<Object>>> it = this.inputResult.asMap().entrySet().iterator();
				while(it.hasNext()) {
					Entry<String, Collection<Object>> entry = it.next();
					String sqlKey = entry.getKey();
					Collection<Object> sqlParams = entry.getValue();
					batchInput(sqlKey, sqlParams);
					it.remove();
					Logger.debug(getClass(), StringCommons.merger("SQL 语句：[", sqlKey, "] 批量执行完成 ..."));
				}
			} else {
				List<Object> sqlParams = this.inputResult.get(sql);
				if (sqlParams.size() >= super.batchSize) {
					batchInput(sql, sqlParams);
					this.inputResult.remove(sql);
					Logger.debug(getClass(), StringCommons.merger("SQL 语句：[", sql, "] 批量执行完成 ..."));
				}
			}
		} catch (Exception e1) {
			if (this.accuratePositioning) {
				Logger.warn(getClass(), "批量执行失败，已开启错误精确定位，使用逐条模式继续执行 ...", e1);
				if (lastBatch) {
					Iterator<Entry<String, Collection<Object>>> it = this.inputResult.asMap().entrySet().iterator();
					while(it.hasNext()) {
						Entry<String, Collection<Object>> entry = it.next();
						String sqlKey = entry.getKey();
						Collection<Object> sqlParams = entry.getValue();
						singleInput(sqlKey, sqlParams);
						it.remove();
					}
				} else {
					List<Object> sqlParams = this.inputResult.get(sql);
					singleInput(sql, sqlParams);
					this.inputResult.remove(sql);
				}
			} else {
				Logger.error(getClass(), "批量执行失败 ...", e1);
				MessageItem<Map<String, Object>> mi = new MessageItem<Map<String, Object>>(null, "非精确定位模式下批量执行失败", Messager.STATUS_ERROR);
				mi.addContent(ExceptionCommons.getStackTrace(e1));
				super.messager.addItem(mi);
			}
		}
	}

	private void batchInput(String sql, Collection<Object> sqlParams) throws Exception {
		if (SQL_RESOURCE.equals(sql)) {
			batchInput(sqlParams.toArray(new String[0]));
		} else {
			batchInput(sql, sqlParams.toArray(new Map[0]));
		}
	}

	private void batchInput(final String[] sqls) throws Exception {
		Exception exception = this.transactionTemplate.execute(new TransactionCallback<Exception>() {
			@Override
			public Exception doInTransaction(TransactionStatus status) {
				Exception exception = null;
				try {
					jdbcTemplate.batchUpdate(sqls);
				} catch (Exception e) {
					status.setRollbackOnly();
					exception = e;
				}
				return exception;
			}
		});
		if (ValidateCommons.isNotEmpty(exception)) {
			throw exception;
		}
	}

	private void batchInput(final String sql, final Map<String, ?>[] batchValues) throws Exception {
		Exception exception = this.transactionTemplate.execute(new TransactionCallback<Exception>() {
			@Override
			public Exception doInTransaction(TransactionStatus status) {
				Exception exception = null;
				try {
					namedParameterJdbcTemplate.batchUpdate(sql, batchValues);
				} catch (Exception e) {
					status.setRollbackOnly();
					exception = e;
				}
				return exception;
			}
		});
		if (ValidateCommons.isNotEmpty(exception)) {
			throw exception;
		}
	}

	private void singleInput(String sql, Collection<Object> sqlParams) {
		for (Object obj : sqlParams) {
			if (SQL_RESOURCE.equals(sql)) {
				String sqlResource = (String) obj;
				try {
					this.jdbcTemplate.update(sqlResource);
					Logger.debug(getClass(), StringCommons.merger("SQL 语句：[", sqlResource, "] 执行完成 ..."));
				} catch (Exception e2) {
					Logger.error(getClass(), StringCommons.merger("SQL 语句：[", sqlResource, "] 执行失败 ..."), e2);
					MessageItem<Map<String, Object>> mi = new MessageItem<Map<String, Object>>(null, StringCommons.merger("SQL 语句：[", sqlResource, "] 执行失败"), Messager.STATUS_ERROR);
					mi.addContent(ExceptionCommons.getStackTrace(e2));
					super.messager.addItem(mi);
				}
			} else {
				Map<String, Object> sqlParam = (Map<String, Object>) obj;
				try {
					this.namedParameterJdbcTemplate.update(sql, sqlParam);
					Logger.debug(getClass(), StringCommons.merger("SQL 语句：[", sql, "] 执行完成，参数列表：[", sqlParam, "] ..."));
				} catch (Exception e2) {
					Logger.error(getClass(), StringCommons.merger("SQL 语句：[", sql, "] 执行失败，参数列表：[", sqlParam, "] ..."), e2);
					MessageItem<Map<String, Object>> mi = new MessageItem<Map<String, Object>>(sqlParam, StringCommons.merger("SQL 语句：[", sql, "] 执行失败"), Messager.STATUS_ERROR);
					mi.addContent(ExceptionCommons.getStackTrace(e2));
					super.messager.addItem(mi);
				}
			}
		}
	}

	@Override
	public void doOutput(String sql, Map<String, Object> params, Callable<Map<String, Object>, Boolean> caller) throws Exception {
		if (ValidateCommons.isNotEmpty(params)) {
			params = new HashMap<String, Object>();
		}
		params.putAll(this.args);

		sql = getSql(sql, params);
		/*
		 * > 基于游标的处理
		 * processCursor(sql, param, caller);
		 * > 基于分页结果集的处理
		 * processResults(sql, param, caller);
		 */
		processCursor(sql, params, caller);
	}

	private void processCursor(String sql, Map<String, Object> params, final Callable<Map<String, Object>, Boolean> caller) {
		// JdbcTemplate 中 resultSetType 默认为 ResultSet.TYPE_FORWARD_ONLY，理论上可以处理大量数据
		this.namedParameterJdbcTemplate.query(sql, params, new RowCallbackHandler() {
			@Override
			public void processRow(ResultSet rs) throws SQLException {
				ResultSetMetaData md = rs.getMetaData();
				int count = md.getColumnCount();

				Map<String, Object> result = new HashMap<String, Object>();
				for (int i = 1; i <= count; i++) {
					String label = md.getColumnLabel(i);
					Object value = rs.getObject(i);
					result.put(label, value);
				}
				result.putAll(params);
				try {
					autoReplace(result);
					caller.call(result);
				} catch (Exception e) {
					throw new SQLException(e);
				}
			}
		});
	}

	private void processResults(String sql, Map<String, Object> params, Callable<Map<String, Object>, Boolean> caller) throws Exception {
		int pageIndex = 1;
		List<Map<String, Object>> results = getOutputResults(sql, params, pageIndex, super.batchSize);
		while (ValidateCommons.isNotEmpty(results)) {
			for (Map<String, Object> result : results) {
				result.putAll(params);
				autoReplace(result);
				caller.call(result);
			}
			pageIndex++;
			results = getOutputResults(sql, params, pageIndex, super.batchSize);
		}
	}

	private List<Map<String, Object>> getOutputResults(String sql, Map<String, Object> params, int pageIndex, int pageLimit) {
		String pagingSql = this.dialectCommons.SQL.buildPagingSql(sql, pageIndex, pageLimit);
		return this.namedParameterJdbcTemplate.queryForList(pagingSql, params);
	}

	private String getSql(String sql, Map<String, Object> params) {
		List<String> innerKeys = PlaceholderCommons.getInnerKeys(sql);
		if (ValidateCommons.isNotEmpty(innerKeys)) {
			String providerFlag = innerKeys.get(0);
			if (EXPRESSION_SQL_PROVIDER.equals(providerFlag)) {
				final String className = innerKeys.get(1);
				String sqlKey = innerKeys.get(2);

				IOSqlProvider provider = null;
				try {
					provider = CacheCommons.get(providerCahce, className, new Callable<String, IOSqlProvider>() {
						@Override
						public IOSqlProvider call(String cacheName) throws Exception {
							IOSqlProvider provider = null;
							try {
								Class<IOSqlProvider> clazz = BeanCommons.getClass(className);
								provider = clazz.newInstance();
							} catch (Exception e) {
								Logger.error(getClass(), StringCommons.merger("未找到或无法实例化 [", className, "]"));
							}
							return provider;
						}
					});
				} catch (Exception e) {
					Logger.error(getClass(), StringCommons.merger("未找到指定的 SqlProvider [", className, "]"), e);
				}
				if (ValidateCommons.isNotEmpty(provider)) {
					sql = provider.getSql(sqlKey, params);
				}
			}
		}
		return sql;
	}

}
